package com.gitee.xhs.keyedState.aggregatingState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.configuration.Configuration;

/**
 * @author BigHead
 * @date 2022/6/30 15:08
 * @description AggregatingStateKeyedProcessFunction
 */
public class AggregatingStateKeyedProcessFunction extends KeyedProcessFunction<Long, Transaction, String> {

    private transient AggregatingState<Transaction, AggregateFunctionDao> aggregatingState;

    @Override
    public void open(Configuration configuration) {
        aggregatingState = getRuntimeContext().getAggregatingState(AggregatingStateConstant.AGGREGATING_STATE_DESCRIPTOR);
    }


    @Override
    public void processElement(Transaction value, KeyedProcessFunction<Long, Transaction, String>.Context ctx, Collector<String> out) throws Exception {
        aggregatingState.add(value);
        out.collect(value.toString());
    }
}
